package com.surfilter.massdata.spark.output.ipcheck;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.Date;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.output.impl.HiveOutput;
import com.surfilter.massdata.spark.util.JdbcUtil;
import com.surfilter.massdata.spark.util.JdbcUtilNew;
import com.surfilter.massdata.spark.util.ipcheck.IpAllocateType;
import com.surfilter.massdata.spark.util.ipcheck.IpCheckUtil;
import com.surfilter.massdata.spark.util.ipcheck.JavaForeachPartitionDFFunc;

import scala.collection.Iterator;

/** 
* @author  hutao
* @date 创建时间：2017年6月9日 上午11:04:16 
* @Description：
* @version 1.0 
* 
**/
public class IpCheckToOracleBr7007 extends HiveOutput {

	private static final long serialVersionUID = 1L;
	private static final Log log = LogFactory.getLog(IpCheckToOracleBr7007.class);
	private String tableName;
	@Override
	public void write(DataFrame result, Map<String, String> commandMap) {
		log.info("===========IpCheckToOracleBr7007 begin============");
		try{
//			IpCheckUtil.truncateTable(tableName);//清空表数据
			System.err.println("br7007 result>>>>>>>>>>>>>>>>>>>>>>>>"+result.count());
			writeToTable(result,tableName);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		log.info("===========IpCheckToOracleBr7007 end============");
	}
	public static void writeToTable(DataFrame df,String tableName) throws Exception {

		final String br7007InsertSql = "INSERT INTO "+tableName+"(ID,BBDW_ID,SBSJ,QSIP,ZZIP,SFGN,FPFS,FPDX,SYFS,FPRQ,YMTZFWQ,DWMC,DWXZ,DWXZJB,DWFL,DWHYFL,JYXKZH,SHENGID,SHIID,XIANID,XXDZ,LXRXM,LXRDH,LXRDZYJ,WGIP,WGDZ,BZ,IPSSS,JTIPLX,IPSSJF,SOURCE_IDC_ID,REPORT_IDC_ID,REPORT_TYPE,LAST_MODIFY_TIME,HOUSE_CODE,IPLX,LYLX,LYID,LYMC)" +
				" values(BR7007_SE.NEXTVAL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
		df.foreachPartition(new JavaForeachPartitionDFFunc() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Iterator<Row> it) {
				// TODO Auto-generated method stub
				Connection conn = null;
				PreparedStatement ps7007 = null;
				boolean status=true;
				try {
					conn = JdbcUtilNew.getConnection();
					conn.setAutoCommit(false);
					int count = 0;
					int batchSize = 5000;
					ps7007 = conn.prepareStatement(br7007InsertSql);
					while (it.hasNext()) {
						Row row = it.next();
						ps7007.setLong(1, Long.parseLong(row.getAs("bbdw_id").toString()));
						ps7007.setDate(2, new java.sql.Date((Long)row.getAs("sbsj")));
						ps7007.setLong(3, Long.parseLong(row.getAs("qsip").toString()));
						ps7007.setLong(4, Long.parseLong(row.getAs("zzip").toString()));
						if(null==row.getAs("sfgn")){
							ps7007.setNull(5, Types.NUMERIC);
						}else{
							ps7007.setLong(5, Long.parseLong(row.getAs("sfgn").toString()));
						}
						Long fpfs=(Long)row.getAs("fpfs");
						ps7007.setLong(6, fpfs);
						if(fpfs.equals(IpAllocateType.RE_ALLOCATE.getTypeValue())){
							if(null==row.getAs("fpdx")){
								ps7007.setNull(7, Types.NUMERIC);
							}else{
								ps7007.setLong(7, Long.parseLong(row.getAs("fpdx").toString()));
							}
							
						}else{
							ps7007.setNull(7, Types.NUMERIC);
						}
						if(null==row.getAs("syfs")){
							ps7007.setNull(8, Types.NUMERIC);
						}else{
							ps7007.setLong(8, Long.parseLong(row.getAs("syfs").toString()));
						}
						if(null==row.getAs("fprq")||((long)row.getAs("fprq"))==-1l){
							ps7007.setNull(9, Types.DATE);
						}else{
							ps7007.setDate(9, new java.sql.Date(Long.parseLong(row.getAs("fprq").toString())));
						}
						if(null==row.getAs("ymtzfwq")){
							ps7007.setNull(10, Types.NUMERIC);
						}else{
							ps7007.setInt(10, Integer.parseInt(row.getAs("ymtzfwq").toString()));
						}
						if(null==row.getAs("dwmc")||((String)row.getAs("dwmc")).equals("N/A")){
							ps7007.setNull(11, Types.VARCHAR);
						}else{
							ps7007.setString(11, row.getAs("dwmc").toString());
						}
						
						if(null==row.getAs("dwxz")){
							ps7007.setNull(12, Types.NUMERIC);
						}else{
							ps7007.setLong(12, Long.parseLong(row.getAs("dwxz").toString()));
						}
						if(null==row.getAs("dwxzjb")){
							ps7007.setNull(13, Types.NUMERIC);
						}else{
							ps7007.setLong(13, Long.parseLong(row.getAs("dwxzjb").toString()));
						}
						if(null==row.getAs("dwfl")){
							ps7007.setNull(14, Types.NUMERIC);
						}else{
							ps7007.setLong(14, Long.parseLong(row.getAs("dwfl").toString()));
						}
						if(null==row.getAs("dwhyfl")){
							ps7007.setNull(15, Types.NUMERIC);
						}else{
							ps7007.setLong(15, Long.parseLong(row.getAs("dwhyfl").toString()));
						}
						if(null==row.getAs("jyxkzh")||((String)row.getAs("jyxkzh")).equals("N/A")){
							ps7007.setNull(16, Types.VARCHAR);
						}else{
							ps7007.setString(16, row.getAs("jyxkzh").toString());
						}
						if(null==row.getAs("shengid")){
							ps7007.setNull(17, Types.NUMERIC);
						}else{
							ps7007.setLong(17, Long.parseLong(row.getAs("shengid").toString()));
						}
						if(null==row.getAs("shiid")){
							ps7007.setNull(18, Types.NUMERIC);
						}else{
							ps7007.setLong(18, Long.parseLong(row.getAs("shiid").toString()));
						}
						if(null==row.getAs("xianid")){
							ps7007.setNull(19, Types.NUMERIC);
						}else{
							ps7007.setLong(19, Long.parseLong(row.getAs("xianid").toString()));
						}
						if(null==row.getAs("xxdz")||((String)row.getAs("xxdz")).equals("N/A")){
							ps7007.setNull(20, Types.VARCHAR);
						}else{
							ps7007.setString(20, row.getAs("xxdz").toString());
						}
						if(null==row.getAs("lxrxm")||((String)row.getAs("lxrxm")).equals("N/A")){
							ps7007.setNull(21, Types.VARCHAR);
						}else{
							ps7007.setString(21, row.getAs("lxrxm").toString());
						}
						if(null==row.getAs("lxrdh")||((String)row.getAs("lxrdh")).equals("N/A")){
							ps7007.setNull(22, Types.VARCHAR);
						}else{
							ps7007.setString(22, row.getAs("lxrdh").toString());
						}
						if(null==row.getAs("lxrdzyj")||((String)row.getAs("lxrdzyj")).equals("N/A")){
							ps7007.setNull(23, Types.VARCHAR);
						}else{
							ps7007.setString(23,row.getAs("lxrdzyj").toString());
						}
						if(null==row.getAs("wgip")){
							ps7007.setNull(24, Types.NUMERIC);
						}else{
							ps7007.setLong(24, Long.parseLong(row.getAs("wgip").toString()));
						}
						if(null==row.getAs("wgdz")||((String)row.getAs("wgdz")).equals("N/A")){
							ps7007.setNull(25, Types.VARCHAR);
						}else{
							ps7007.setString(25, row.getAs("wgdz").toString());
						}
						if(null==row.getAs("bz")||((String)row.getAs("bz")).equals("N/A")){
							ps7007.setNull(26, Types.VARCHAR);
						}else{
							ps7007.setString(26, row.getAs("bz").toString());
						}
						if(null==row.getAs("ipsss")){
							ps7007.setNull(27, Types.NUMERIC);
						}else{
							ps7007.setLong(27, Long.parseLong(row.getAs("ipsss").toString()));
						}
						if(null==row.getAs("jtiplx")){
							ps7007.setNull(28, Types.NUMERIC);
						}else{
							ps7007.setInt(28, Integer.parseInt(row.getAs("jtiplx").toString()));
						}
						if(null==row.getAs("ipssjf")||((String)row.getAs("ipssjf")).equals("N/A")){
							ps7007.setNull(29, Types.VARCHAR);
						}else{
							ps7007.setString(29, row.getAs("ipssjf").toString());
						}
						if(null==row.getAs("sourceIdcId") ||((String)row.getAs("sourceIdcId")).equals("N/A")){
							ps7007.setNull(30, Types.VARCHAR);
						}else{
							ps7007.setString(30, row.getAs("sourceIdcId").toString());
						}
						if(null==row.getAs("reportIdcId")||((String)row.getAs("reportIdcId")).equals("N/A")){
							ps7007.setNull(31, Types.VARCHAR);
						}else{
							ps7007.setString(31,row.getAs("reportIdcId").toString());
						}
						if(row.getAs("reportType")== null){
							ps7007.setInt(32, 0);
						}else{
							ps7007.setInt(32, Integer.parseInt(row.getAs("reportType").toString()));
						}
						if(null==row.getAs("lastModifyTime")){
							ps7007.setDate(33, new java.sql.Date(new Date().getTime()));
						}else{
							ps7007.setDate(33, new java.sql.Date(Long.parseLong(row.getAs("lastModifyTime").toString())));
						}
						
						if(null==row.getAs("houseCode")||((String)row.getAs("houseCode")).equals("N/A")){
							ps7007.setNull(34, Types.VARCHAR);
						}else{
							ps7007.setString(34, row.getAs("houseCode").toString());
						}
						
						if(null==row.getAs("iplx")){
							ps7007.setNull(35, Types.NUMERIC);
						}else{
							ps7007.setInt(35, Integer.parseInt(row.getAs("iplx").toString()));
						}
						if(null==row.getAs("lylx") || (Integer)row.getAs("lylx") == -1l){
							ps7007.setNull(36, Types.NUMERIC);
						}else{
							ps7007.setInt(36, Integer.parseInt(row.getAs("lylx").toString()));
						}
						if(null==row.getAs("fpLyId") || (Long)row.getAs("fpLyId") == -1l){
							ps7007.setNull(37, Types.NUMERIC);
						}else{
							ps7007.setLong(37, Long.parseLong(row.getAs("fpLyId").toString()));
						}
						if(null==row.getAs("lymc")||((String)row.getAs("lymc")).equals("N/A")){
							ps7007.setNull(38, Types.VARCHAR);
						}else{
							ps7007.setString(38, row.getAs("lymc").toString());
						}
						
						ps7007.addBatch();
						if(++count % batchSize==0){
							ps7007.executeBatch();
							conn.commit();
							ps7007.clearBatch();
							count = 0;
						}
						
					}//end while
//					if(count > 0){
//						ps7007.executeBatch();
//						ps7007.clearBatch();
//						conn.commit();
//						count = 0;
//					}
					ps7007.executeBatch();
					conn.commit();
					ps7007.clearBatch();
					
				} catch (Exception e) {
					e.printStackTrace();
					
				} finally {
					if (null != ps7007) {
						try {
							ps7007.close();
						} catch (SQLException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
					if (conn != null) {
						try {
							conn.close();
						} catch (SQLException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				}

			}
		});

	}
}